package com.ibm.cloud.redismq.mq;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.ibm.cloud.redismq.conf.RedisMQProperties;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class MQManager {
	@Autowired RedisMQProperties properties;
	@Autowired MessageListener listener;
	@Autowired MessageRecoverier recoverier;
	
	private boolean stopped = true;
	private List<Thread> workers = new ArrayList<>();
	
	@PostConstruct
	public void start() {
		if (!stopped) {
			return;
		}
		stopped = false;
		log.info("Starting Workers...");
		for (int i = 0; i < properties.getConsumerCount(); i++) {
			Thread lisThread = new Thread(listener);
			lisThread.setName(listener.getClass().getSimpleName() + i);
			lisThread.start();
			workers.add(lisThread);
		}
		log.info("{} MessageListeners started", properties.getConsumerCount());
		for (int i = 0; i < properties.getRecoverierCount(); i++) {
			Thread recThread = new Thread(recoverier);
			recThread.setName(recoverier.getClass().getSimpleName() + i);
			recThread.start();
			workers.add(recThread);
		}
		log.info("{} MessageRecoveriers started.", properties.getRecoverierCount());
	}
	@PreDestroy
	public void stop() {
		if (stopped) {
			log.warn("Already stopped or is stopping...");
			return;
		}
		log.info("Workers are stopping...");	
		listener.setStarted(false);
		recoverier.setStarted(false);
		// interrupt the worker thread so that it can terminate
		for (Thread worker : workers) {
			worker.interrupt();
		}
		for (Thread worker : workers) {
			try {
				worker.join(properties.getMaxFlushTime());

				// check to see if the thread ended and if not add a warning message
				if (worker.isAlive()) {
					log.warn("Max queue flush timeout (" + properties.getMaxFlushTime() + " ms) exceeded. Approximately ");
				} else {
					log.info("Queue flush finished successfully for:{}" + worker.getName());
				}
			} catch (InterruptedException e) {
				log.error("Failed to join worker thread {}", worker.getName(), e);
			}
		}

	}
}
